package com.dtruth.dataflow.bolt;

import java.math.BigInteger;
import java.util.Map;
import java.util.Set;

import org.apache.thrift.TException;
import org.slf4j.Logger;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import com.dtruth.dataflow.dao.RedisDAO;
import com.dtruth.dataflow.util.ETLServiceClient;
import com.dtruth.dataflow.util.Env;
import com.dtruth.dataflow.util.HtmlRegexpUtil;
import com.dtruth.dataflow.util.MyLoggerFactory;
import com.dtruth.dataflow.util.SimHashUtil;
import com.dtruth.dataflow.util.StringUtil;
import com.feheadline.cms.general.etl.service.thrift.gen.ETLThriftService.Client;

/**
 * 频道自动划分，分类提取
 * 
 * @author 岳海亮
 * @email yhl@feheadline.com
 * @date 2014年10月27日
 */
public class RmDuplctBolt extends BaseBasicBolt {

	private static final long serialVersionUID = 1L;

	private static Logger log = MyLoggerFactory.getLogger(RmDuplctBolt.class);

	RedisDAO rd = null;
	String domain = null;

	/**
	 * On create
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		String host = Env.getInstance().getProperty("redis.rmduplct.host")
				.trim();
		int port = Integer.valueOf(Env.getInstance()
				.getProperty("redis.rmduplct.port").trim());
		int insNum = Integer.valueOf(Env.getInstance()
				.getProperty("redis.rmduplct.db").trim());
		rd = new RedisDAO(host, port, insNum);
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		domain = map.get("domain").toString();
		String title = map.get("title").toString();
		String body = map.get("body").toString();
		String id = map.get("key").toString();
		String url = map.get("url").toString();
		String quality = map.get("quality") == null ? "5" : map.get("quality")
				.toString();
		BigInteger bi_content = SimHashUtil.getSimHash(HtmlRegexpUtil
				.getValidContent(body));
		BigInteger bi_title = SimHashUtil.getSimHash(title);
		if (isDuplctWithTmp(id, bi_title, bi_content, quality))
			return;
		rd.hset("tmpnewskey:" + id, "content", bi_content.toString());
		rd.hset("tmpnewskey:" + id, "title", bi_title.toString());
		rd.hset("tmpnewskey:" + id, "quality", quality);
		if (!isDuplct(url, id, bi_title, bi_content)) {
			rd.saddWithTTL("title:" + bi_title, id);
			rd.setWithTTL("content:" + bi_content, id);
			rd.hincrOrHset("newskey:" + id, bi_title.toString(),
					bi_content.toString(), quality);
			log.debug("emit:" + map.get("url"));
			collector.emit(new Values(map));
		}
		rd.del("tmpnewskey:" + id);
	}

	private boolean isDuplctWithTmp(String id, BigInteger bi_title,
			BigInteger bi_content, String quality) {
		Set<String> set = rd.getKeys("tmpnewskey:*");
		if (set == null || set.size() == 0)
			return false;
		for (String newskey : set) {
			String _content = rd.hget(newskey, "content");
			String _title = rd.hget(newskey, "title");
			if (_content == null || !StringUtil.isNumeric(_content))
				continue;
			int contentDis = SimHashUtil.hammingDistance(bi_content,
					new BigInteger(_content));
			if (contentDis < 2) {
				log.info(id + "(" + domain + ")" + "=" + newskey
						+ ":contentDis=" + contentDis);
				rd.hincrOrHset4Tmp(newskey, quality);
				return true;
			}
			int titleDis = 5;
			if (_title != null && StringUtil.isNumeric(_title))
				titleDis = SimHashUtil.hammingDistance(bi_title,
						new BigInteger(_title));
			if (titleDis < 3 && contentDis < 5) {
				log.info(id + "(" + domain + ")" + "=" + newskey
						+ ":contentDis=" + contentDis + ",titleDis ="
						+ titleDis);
				rd.hincrOrHset4Tmp(newskey, quality);
				return true;
			}
		}
		return false;
	}

	/**
	 * 
	 * @param id
	 * @param bi_title
	 * @param bi_content
	 * @return
	 */
	private boolean isDuplct(String url, String id, BigInteger bi_title,
			BigInteger bi_content) {
		// 判断内容是否相同
		String key = rd.get("content" + ":" + bi_content);
		if (key != null) {
			if (key.equals(id)) {
				log.info("the same url :" + id + "(" + domain + ")" + "=" + key);
			} else {
				log.info("content is total same :" + id + "(" + domain + ")"
						+ "=" + key);
				String quality = rd.hget("tmpnewskey:" + id, "quality");
				if (updateNewsScore(url, key, quality))
					rd.hincrOrHset("newskey:" + key,
							rd.hget("newskey:" + key, "title"),
							bi_content.toString(), quality);
			}
			return true;
		}

		Set<String> keys = rd.smembers("title:" + bi_title);// 取得标题相同的新闻newskey
		if (keys != null && keys.size() > 0) {
			for (String _key : keys) {
				String value = rd.hget("newskey:" + _key, "content");
				if (value == null)
					continue;
				int dis = SimHashUtil.hammingDistance(bi_content,
						new BigInteger(value));
				if (dis < 8) {// 标题相同时，判断内容距离
					log.info("title is same and " + id + "(" + domain + ")"
							+ "=" + _key + ":" + dis);
					String quality = rd.hget("tmpnewskey:" + id, "quality");
					if (updateNewsScore(url, _key, quality))
						rd.hincrOrHset("newskey:" + key,
								rd.hget("newskey:" + key, "title"), value,
								quality);
					return true;
				}
			}
		}
		Set<String> set = rd.getKeys("newskey:*");
		for (String newskey : set) {
			String _content = rd.hget(newskey, "content");
			String _title = rd.hget(newskey, "title");
			if (_content == null || !StringUtil.isNumeric(_content))
				continue;
			int contentDis = SimHashUtil.hammingDistance(bi_content,
					new BigInteger(_content));
			if (contentDis < 2) {
				log.info(id + "(" + domain + ")" + "=" + newskey
						+ ":contentDis=" + contentDis);
				String quality = rd.hget("tmpnewskey:" + id, "quality");
				if (updateNewsScore(url,
						newskey.substring(newskey.indexOf(":") + 1), quality))
					rd.hincrOrHset(newskey, _title, _content, quality);
				return true;
			}
			int titleDis = 5;
			if (_title != null && StringUtil.isNumeric(_title))
				titleDis = SimHashUtil.hammingDistance(bi_title,
						new BigInteger(_title));
			if (titleDis < 3 && contentDis < 5) {
				log.info(id + "(" + domain + ")" + "=" + newskey
						+ ":contentDis=" + contentDis + ",titleDis ="
						+ titleDis);
				String quality = rd.hget("tmpnewskey:" + id, "quality");
				if (updateNewsScore(url,
						newskey.substring(newskey.indexOf(":") + 1), quality))
					rd.hincrOrHset(newskey, _title, _content, quality);
				return true;
			}

		}
		return false;
	}

	/**
	 * 通过接口更新数据库中的新闻质量分
	 * 
	 * @param key
	 * @param quality
	 * @return
	 */
	private boolean updateNewsScore(String url, String key, String quality) {
		ETLServiceClient esc = new ETLServiceClient();
		String address = Env.getInstance().getProperty("data.pushout.url");
		Client client = null;
		try {
			client = esc.getClient(address);
			log.debug("update score by:" + url);
			return client.updateNewsScore(key, Double.parseDouble(quality));
		} catch (TException e) {
			log.warn("updateNewsScore:" + key + "(" + domain + ")");
		} finally {
			if (client != null)
				esc.closeClient();
		}
		return false;
	}
}